/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.io.File;
import java.nio.ByteBuffer;
import java.util.logging.Logger;

import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.io.FileReference;
import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
import edu.uci.ics.hyracks.storage.common.buffercache.ICachedPage;
import edu.uci.ics.hyracks.storage.common.file.BufferedFileHandle;
import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
import edu.ucsd.db.bassarisk.data.api.IValueType;
import edu.ucsd.db.bassarisk.storage.columns.IColumnFactory;

public class CabinetReader implements IDrawerBufferSource {
  private static final Logger LOG = Logger.getLogger(CabinetReader.class.getName());
  
  private final IBufferCache bufferCache;
  private final ILexiconReader lexReader;
  private final int cabinetFileId;
  
  public CabinetReader(IBufferCache bufferCache, String bufferFileName,
      IFileMapProvider fileMapProvider, ILexiconReader lexicon)
      throws HyracksDataException {
    // assign injected dependences
    this.bufferCache = bufferCache;
    this.lexReader = lexicon;
    
    // Open cabinet file within buffer cache
    FileReference file = new FileReference(new File(bufferFileName));
    bufferCache.createFile(file);
    this.cabinetFileId = fileMapProvider.lookupFileId(file);
    bufferCache.openFile(this.cabinetFileId);
  }
  
  public ColumnReader<?> openColumn(String key) {
    int columnHeaderAddress = this.lexReader.findColumn(key);
    
    return this.openColumn(columnHeaderAddress, null);
  }
  
  public <VT extends IValueType> ColumnReader<VT> openColumn(int columnHeadDrawer,
                                                             IColumnFactory<VT> columnFactory) {
    ColumnReader<VT> reader = columnFactory.makeColumnReader(this, columnHeadDrawer);
    return reader;
  }
  
  public DrawerBuffer makeDrawerBuffer() {
    return new DrawerBuffer();
  }
  
  @Override
  public void setDrawer(int columnHeadDrawer, DrawerBuffer drawerBuffer) {
//    LOG.info("Loading buffer #" + columnHeadDrawer);
    long dpid = BufferedFileHandle.getDiskPageId(this.cabinetFileId, columnHeadDrawer);
    try {
      ICachedPage drawerPage = this.bufferCache.pin(dpid, false);
      drawerPage.acquireReadLatch();
      drawerBuffer.reset(drawerPage);
    } catch (HyracksDataException e) {
      throw new RuntimeException(e);
    }
  }
  
  public void closeColumn(ColumnReader<?> reader) {
    reader.close();
  }
  
  public void close() {
    this.lexReader.close();
  }
  
  protected class DrawerBuffer {
    private ICachedPage drawerPage;
    
    private DrawerBuffer() {}
    
    private void reset(ICachedPage drawerPage) {
      if (this.drawerPage != null) {
        close();
      }
      this.drawerPage = drawerPage;
      this.drawerPage.acquireReadLatch();
    }
    
    public ByteBuffer getBuffer() {
      return this.drawerPage.getBuffer();
    }
    
    public void close() {
      this.drawerPage.releaseReadLatch();
      try {
        bufferCache.unpin(this.drawerPage);
      } catch (HyracksDataException e) {
        throw new RuntimeException(e); // naughty naughty
      }
      this.drawerPage = null;
    }
  }
}
